package cn.enjoy.rabbitmq.config;

import cn.hutool.core.date.DateUtil;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Map;


/**
 * 重试后，消费失败统一处理
 *
 * @author K
 * @date 2022-07-20 15:18
 */
@Slf4j
public class MqRepublishMessageRecoverer extends RepublishMessageRecoverer {

    private final static String X_EXCEPTION_TIME = "x-exception-time";
    private final AmqpTemplate errorTemplate;
    private final String errorRoutingKey;
    private final String errorExchangeName;
    private volatile String errorRoutingKeyPrefix = "error.";

    public MqRepublishMessageRecoverer(AmqpTemplate errorTemplate, String errorRoutingKey, String errorExchangeName) {
        super(errorTemplate);
        this.errorTemplate = errorTemplate;
        this.errorRoutingKey = errorRoutingKey;
        this.errorExchangeName = errorExchangeName;
    }

    @Override
    public void recover(Message message, Throwable cause) {

        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        headers.put(X_EXCEPTION_STACKTRACE, getStackTraceAsString(cause));
        headers.put(X_EXCEPTION_MESSAGE, cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage());
        headers.put(X_ORIGINAL_EXCHANGE, message.getMessageProperties().getReceivedExchange());
        headers.put(X_ORIGINAL_ROUTING_KEY, message.getMessageProperties().getReceivedRoutingKey());
        headers.put(X_EXCEPTION_TIME, DateUtil.now());
        Map<? extends String, ? extends Object> additionalHeaders = additionalHeaders(message, cause);
        if (additionalHeaders != null) {
            headers.putAll(additionalHeaders);
        }
        log.error("mq recover more times, exception=>", cause);
        if (null != this.errorExchangeName) {
            errorTemplate.send(this.errorExchangeName, this.errorRoutingKey, message);
            log.error("Republishing failed message to exchange " + this.errorExchangeName);
        } else {
            final String routingKey = this.prefixedOriginalRoutingKey(message);
            errorTemplate.send(this.errorRoutingKey, message);
            log.error("Republishing failed message to the template's default exchange with routing key " + routingKey);
        }
    }

    private String getStackTraceAsString(Throwable cause) {
        StringWriter stringWriter = new StringWriter();
        PrintWriter printWriter = new PrintWriter(stringWriter, true);
        cause.printStackTrace(printWriter);
        return stringWriter.getBuffer().toString();
    }

    private String prefixedOriginalRoutingKey(Message message) {
        return this.errorRoutingKeyPrefix + message.getMessageProperties().getReceivedRoutingKey();
    }

    @Override
    protected Map<? extends String, ? extends Object> additionalHeaders(Message message, Throwable cause) {
        Map<String, Object> map = Maps.newHashMap();
        map.put("content_type", "application/json");
        map.put("content_encoding", "UTF-8");
        return map;
    }

}
